package com.xy.reactor.demos;


import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * 流的异常演示
 */
public class ErrorReactorDemo {

    /**
     * onErrorStop  发生异常直接关闭当前流。  截断的概念
     */
    @Test
    public void catchStop(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .onErrorStop()
                .doFinally(item ->{
                    // 一定会执行
                    System.out.println("doFinally = " + item);

                })
                .subscribe(item -> {

                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常 进行执行
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 不执行
                            System.out.println("completeConsumer");
                        }
                );

        //subscribe1
        //subscribe2
        //subscribe3
        //subscribe4
        //errorjava.lang.ArithmeticException: / by zero
        //doFinally = onError
    }


    /**
     * 忽略当前异常，进行业务记录,然后再次执行
     */
    @Test
    public void catchFinallyContinue(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .onErrorContinue((error,val) -> {
                    // 可做些内部操作处理
                    System.out.println("err已被记录 = " + error);
                    System.out.println("err的值 = " + val);

                })
                .doFinally(item ->{
                    // 一定会执行
                    System.out.println("doFinally = " + item);

                })
                .subscribe(item -> {

                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常 进行执行
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 执行
                            System.out.println("completeConsumer");
                        }
                );
        // output
//subscribe1
//subscribe2
//subscribe3
//subscribe4
//err已被记录 = java.lang.ArithmeticException: / by zero
//err的值 = 5
//err已被记录 = java.lang.ArithmeticException: / by zero
//err的值 = 6
//err已被记录 = java.lang.ArithmeticException: / by zero
//err的值 = 7
//err已被记录 = java.lang.ArithmeticException: / by zero
//err的值 = 8
//err已被记录 = java.lang.ArithmeticException: / by zero
//err的值 = 9
//err已被记录 = java.lang.ArithmeticException: / by zero
//err的值 = 10
//completeConsumer
//doFinally = onComplete
//Disconnected from the target VM, address: '127.0.0.1:51023', transport: 'socket'
//
//Process finished with exit code 0

    }





    /**
     * Use the finally block to clean up resources or a Java 7 “try-with-resource” construct.
     *
     * try catch 类似逻辑。 会有一个最终执行方法
     */
    @Test
    public void catchFinallyException(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .doOnError(error -> {
                    // 可做些内部操作处理
                    System.out.println("err已被记录 = " + error);
                })
                .doFinally(item ->{
                    // 一定会执行
                    System.out.println("doFinally = " + item);

                })
                .subscribe(item -> {

                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常 进行执行
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 不执行
                            System.out.println("completeConsumer");
                        }
                );
        // output
//subscribe1
//subscribe2
//subscribe3
//subscribe4
//err已被记录 = java.lang.ArithmeticException: / by zero
//errorjava.lang.ArithmeticException: / by zero
//doFinally = onError

    }




    /**
     * Catch, log an error-specific message, and re-throw.
     * 捕获异常，记录特殊的错误日志，重新抛出
     * completeConsumer 不执行
     */
    @Test
    public void catchLogBusinessException(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .doOnError(error -> {
                    // 可做些内部操作处理
                    System.out.println("err已被记录 = " + error);
                })
                .subscribe(item -> {

                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常 进行执行
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 不执行
                            System.out.println("completeConsumer");
                        }
                );
        // output
//subscribe1
//subscribe2
//subscribe3
//subscribe4
//err已被记录 = java.lang.ArithmeticException: / by zero
//errorjava.lang.ArithmeticException: / by zero
    }

    /**
     * Catch, wrap to a BusinessException, and re-throw.
     * 捕获并包装成一个业务异常，并重新抛出
     *  将流截断。completeConsumer 不进行执行。
     */
    @Test
    public void catchBusinessException(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .onErrorMap(error -> {
                    // 可做些内部操作处理
                    return new BuiException(error.getMessage());
                })
                .subscribe(item -> {

                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常 进行执行
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 不执行
                            System.out.println("completeConsumer");
                        }
                );
        // output
/**
 * subscribe1
 * subscribe2
 * subscribe3
 * subscribe4
 * errorcom.xy.reactor.demos.BuiException: / by zero
 */

    }



    /**
     * Catch and dynamically compute a fallback value.
     * 捕获并动态计算一个返回值
     *  onErrorResume 进行 再次抛出异常，这个操作能够让直接订阅者 的异常感知到
     *                 截断操作，将流直接截断了。
     */

    @Test
    public void catchDynamically(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .onErrorResume(error -> {
                          // 可做些内部操作处理
                    return Flux.error(new BuiException(error.getMessage()));
                })
                .subscribe(item -> {

                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常 进行执行
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 不执行
                            System.out.println("completeConsumer");
                        }
                );
        // output
/**
 * subscribe1
 * subscribe2
 * subscribe3
 * subscribe4
 * errorcom.xy.reactor.demos.BuiException: / by zero
 */
    }

    /**
     * Catch and execute an alternative path with a fallback method.
     * 吃掉异常，执行一个兜底方法（发布者类型方法）；
     */
    @Test
    public void catchFallBack(){
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .onErrorResume(error -> {
                    System.err.println("error"+error);
                    // 返回这个发布者方法,就会走此方法。
                    return Mono.just(-100);
                })
                .subscribe(item -> {
                            // 输出 -100
                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常不执行的
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 执行的
                            System.out.println("completeConsumer");
                        }
                );
        // output
//subscribe1
//subscribe2
//subscribe3
//subscribe4
//errorjava.lang.ArithmeticException: / by zero
//subscribe-100
//completeConsumer
    }


    /**
     * Catch and return a static default value
     * 捕获异常返回一个静态默认值 然后结束整个流。
     */
    @Test
    public void catchReturn() {
        Flux.range(1, 10)
                .map(item -> {
                    if (item >=5)return item / 0;
                    return item;
                })
                .onErrorReturn(ArithmeticException.class, 668)
                .subscribe(item -> {
                            // 输出 668
                            System.out.println("subscribe" + item);
                        }, error -> {
                            // 此订阅者异常不执行的
                            System.err.println("error" + error);
                        },
                        () -> {
                            // 执行的
                            System.out.println("completeConsumer");
                        }
                );
        // output
        //subscribe1
        //subscribe2
        //subscribe3
        //subscribe4
        //subscribe668
        //completeConsumer

    }
}


class BuiException extends RuntimeException{
    public BuiException(String message){
        super(message);
    }
}